package com.ruoyi.system.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DateUtil;
import com.mysql.cj.protocol.Resultset;
import com.ruoyi.common.core.exception.ServiceException;
import com.ruoyi.common.core.utils.StringUtils;
import com.ruoyi.common.mybatis.core.page.PageQuery;
import com.ruoyi.common.mybatis.core.page.TableDataInfo;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.ruoyi.common.redis.utils.RedisUtils;
import com.ruoyi.system.domain.*;
import com.ruoyi.system.enums.ArchiveJobExecuteStatusEnum;
import com.ruoyi.system.enums.ArchiveLocationTypeEnum;
import com.ruoyi.system.enums.ArchiveStrategyTypeEnum;
import com.ruoyi.system.mapper.*;
import com.ruoyi.system.util.ArchiveUtil;
import com.ruoyi.system.util.Constants;
import com.ruoyi.system.util.NetUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import com.ruoyi.system.domain.bo.ArchiveRecoverTaskBo;
import com.ruoyi.system.domain.vo.ArchiveRecoverTaskVo;
import com.ruoyi.system.service.IArchiveRecoverTaskService;

import java.io.*;
import java.sql.*;
import java.time.Duration;
import java.util.*;
import java.util.Date;

/**
 * 归档恢复任务Service业务层处理
 *
 * @author ruoyi
 * @date 2024-12-16
 */
@Slf4j
@RequiredArgsConstructor
@Service
public class ArchiveRecoverTaskServiceImpl implements IArchiveRecoverTaskService {

    private final ArchiveRecoverTaskMapper baseMapper;

    private final ArchiveResultMapper resultMapper;

    private final ArchiveStrategyMapper strategyMapper;

    private final ArchiveTaskMapper archiveTaskMapper;

    private final ArchiveLocationMapper locationMapper;

    /**
     * 查询归档恢复任务
     */
    @Override
    public ArchiveRecoverTaskVo queryVoById(Long id) {
        return baseMapper.selectVoById(id);
    }

    @Override
    public ArchiveRecoverTask queryById(Long id) {
        return baseMapper.selectById(id);
    }

    /**
     * 查询归档恢复任务列表
     */
    @Override
    public TableDataInfo<ArchiveRecoverTaskVo> queryPageList(ArchiveRecoverTaskBo bo, PageQuery pageQuery) {
        LambdaQueryWrapper<ArchiveRecoverTask> lqw = buildQueryWrapper(bo);
        Page<ArchiveRecoverTaskVo> result = baseMapper.selectVoPage(pageQuery.build(), lqw);
        return TableDataInfo.build(result);
    }

    /**
     * 查询归档恢复任务列表
     */
    @Override
    public List<ArchiveRecoverTaskVo> queryList(ArchiveRecoverTaskBo bo) {
        LambdaQueryWrapper<ArchiveRecoverTask> lqw = buildQueryWrapper(bo);
        return baseMapper.selectVoList(lqw);
    }

    private LambdaQueryWrapper<ArchiveRecoverTask> buildQueryWrapper(ArchiveRecoverTaskBo bo) {
        Map<String, Object> params = bo.getParams();
        LambdaQueryWrapper<ArchiveRecoverTask> lqw = Wrappers.lambdaQuery();
        lqw.like(StringUtils.isNotBlank(bo.getName()), ArchiveRecoverTask::getName, bo.getName());
        lqw.like(StringUtils.isNotBlank(bo.getTableName()), ArchiveRecoverTask::getTableName, bo.getTableName());
        lqw.eq(bo.getResultId() != null, ArchiveRecoverTask::getResultId, bo.getResultId());
        lqw.eq(bo.getRecoveryType() != null, ArchiveRecoverTask::getRecoveryType, bo.getRecoveryType());
        lqw.eq(bo.getExecuteStatus() != null, ArchiveRecoverTask::getExecuteStatus, bo.getExecuteStatus());
        return lqw;
    }

    /**
     * 新增归档恢复任务
     */
    @Override
    public Boolean insertByBo(ArchiveRecoverTaskBo bo) {
        ArchiveRecoverTask add = BeanUtil.toBean(bo, ArchiveRecoverTask.class);
        add.setExecuteStatus(ArchiveJobExecuteStatusEnum.WAIT_EXECUTE.getValue());
        validEntityBeforeSave(add);
        boolean flag = baseMapper.insert(add) > 0;
        if (flag) {
            bo.setId(add.getId());
        }
        return flag;
    }

    /**
     * 修改归档恢复任务
     */
    @Override
    public Boolean updateByBo(ArchiveRecoverTaskBo bo) {
        ArchiveRecoverTask update = BeanUtil.toBean(bo, ArchiveRecoverTask.class);
        update.setExecuteStatus(ArchiveJobExecuteStatusEnum.WAIT_EXECUTE.getValue());
        validEntityBeforeUpdate(update);
        validEntityBeforeSave(update);
        return baseMapper.updateById(update) > 0;
    }

    /**
     * 保存前的数据校验
     */
    private void validEntityBeforeSave(ArchiveRecoverTask entity) {
        ArchiveResult result = resultMapper.selectById(entity.getResultId());
        if (result == null) {
            throw new ServiceException("归档结果不存在");
        }
        if (result.getExecuteStatus() != ArchiveJobExecuteStatusEnum.SUCCESS.getValue()) {
            throw new ServiceException("归档结果未成功执行，不能进行恢复");
        }
        entity.setResultName(result.getTaskName() + "-" + result.getResultNo());
    }

    /**
     * 保存前的数据校验
     */
    private void validEntityBeforeUpdate(ArchiveRecoverTask entity) {
        ArchiveRecoverTask task = baseMapper.selectById(entity.getId());
        if (task == null) {
            throw new ServiceException("恢复任务不存在");
        }
        if (ArchiveJobExecuteStatusEnum.EXECUTING.getValue() == task.getExecuteStatus()) {
            throw new ServiceException("恢复任务执行中，不能进行操作");
        }
    }

    /**
     * 批量删除归档恢复任务
     */
    @Override
    public Boolean deleteWithValidByIds(Collection<Long> ids, Boolean isValid) {
        if (isValid) {
            List<ArchiveRecoverTask> list = baseMapper.selectBatchIds(ids);
            list.forEach(task -> {
                if (ArchiveJobExecuteStatusEnum.EXECUTING.getValue() == task.getExecuteStatus()) {
                    throw new ServiceException(String.format("%s 恢复任务执行中，不能进行操作", task.getName()));
                }
            });
        }
        return baseMapper.deleteBatchIds(ids) > 0;
    }

    @Override
    public Boolean execute(Long id) {
        ArchiveRecoverTask task = baseMapper.selectById(id);
        if (task == null) {
            throw new ServiceException("恢复任务不存在");
        }
        if (ArchiveJobExecuteStatusEnum.EXECUTING.getValue() == task.getExecuteStatus()) {
            throw new ServiceException("恢复任务执行中，不能进行操作");
        }

        execute(task);

        return Boolean.TRUE;
    }

    @Async
    protected void execute(ArchiveRecoverTask task) {
        String key = String.format("ArchiveRecoverTask:%d", task.getId());
        boolean lock = RedisUtils.setObjectIfAbsent(key, task.getId(), Duration.ofHours(1));
        if (!lock) {
            return;
        }
        try {
            String msg = String.format("%s：开始执行", DateUtil.formatDateTime(new Date()));
            baseMapper.updateStartInfo(task.getId(), msg);
            recoverCheck(task);
            doRecover(task);
            baseMapper.updateFinishInfo(task.getId(), ArchiveJobExecuteStatusEnum.SUCCESS.getValue(), ArchiveUtil.getInfo("恢复完成"));
        } catch (Exception e) {
            baseMapper.updateFinishInfo(task.getId(), ArchiveJobExecuteStatusEnum.FAILED.getValue(), ArchiveUtil.getInfo(e.getMessage()));
        } finally {
            RedisUtils.deleteObject(key);
        }
    }

    protected void doRecover(ArchiveRecoverTask task) throws Exception {
        ArchiveStrategy archiveStrategy = task.getArchiveStrategy();
        if (ArchiveStrategyTypeEnum.COMPRESS_TABLE.getValue() == archiveStrategy.getStrategyType()
            && ArchiveLocationTypeEnum.DB.getValue() == task.getArchiveLocation().getLocationType()) {
            recoverFromDb(task);
        } else if (ArchiveStrategyTypeEnum.SQL_FILE.getValue() == archiveStrategy.getStrategyType()) {
            if (ArchiveLocationTypeEnum.LOCAL.getValue() == task.getArchiveLocation().getLocationType()) {
                recoverFromSqlLocal(task);
            } else if (ArchiveLocationTypeEnum.S3.getValue() == task.getArchiveLocation().getLocationType()) {
                recoverFromSqlS3(task);
            } else {
                throw new ServiceException(String.format("%s：恢复失败，要恢复的归档位置错误", DateUtil.formatDateTime(new Date())));
            }
        } else {
            throw new ServiceException(String.format("%s：恢复失败，要恢复的归档策略类型错误", DateUtil.formatDateTime(new Date())));
        }
    }

    private void recoverFromSqlS3(ArchiveRecoverTask recoverTask) throws Exception {
        ArchiveTask archiveTask = recoverTask.getArchiveTask();
        ArchiveResult archiveResult = recoverTask.getArchiveResult();
        ArchiveLocation archiveLocation = recoverTask.getArchiveLocation();
        String s3Url = ArchiveUtil.getS3FilePath(archiveResult, archiveLocation);
        String localFilePath = String.format("/tmp/archive/%d/%s.sql", archiveTask.getId(), archiveResult.getResultNo());
        baseMapper.updateFinishInfo(archiveTask.getId(), ArchiveJobExecuteStatusEnum.EXECUTING.getValue(),
            ArchiveUtil.getInfo(String.format("开始从对象存储下载归档文件：%s 到本地临时文件：%s", s3Url, localFilePath)));
        NetUtil.downloadFile(s3Url, localFilePath);
        baseMapper.updateFinishInfo(archiveTask.getId(), ArchiveJobExecuteStatusEnum.EXECUTING.getValue(),
            ArchiveUtil.getInfo(String.format("从对象存储下载归档文件：%s 到本地临时文件：%s 成功", s3Url, localFilePath)));

        File sqlFile = new File(localFilePath);
        if (!sqlFile.exists()) {
            throw new ServiceException(String.format("%s：恢复失败，要恢复的本地临时SQL文件不存在", DateUtil.formatDateTime(new Date())));
        }
        recoverFromSqlLocal(recoverTask, archiveResult, sqlFile);
        sqlFile.delete();
        baseMapper.updateFinishInfo(archiveTask.getId(), ArchiveJobExecuteStatusEnum.EXECUTING.getValue(),
            ArchiveUtil.getInfo(String.format("删除本地临时文件：%s 成功", s3Url, localFilePath)));
    }

    private void recoverFromSqlLocal(ArchiveRecoverTask recoverTask) throws Exception {
        ArchiveTask archiveTask = recoverTask.getArchiveTask();
        ArchiveResult result = recoverTask.getArchiveResult();
        ArchiveLocation location = recoverTask.getArchiveLocation();
        File sqlFile = new File(String.format("%s/archive/%d/%s.sql", location.getLocalDir(), archiveTask.getId(), result.getResultNo()));
        if (!sqlFile.exists()) {
            throw new ServiceException(String.format("%s：恢复失败，要恢复的SQL文件不存在", DateUtil.formatDateTime(new Date())));
        }
        recoverFromSqlLocal(recoverTask, result, sqlFile);
    }

    private void recoverFromSqlLocal(ArchiveRecoverTask recoverTask, ArchiveResult result, File sqlFile) throws Exception {
        long totalCount = result.getTotalCount();
        long index = 0;
        try (BufferedReader reader = new BufferedReader(new FileReader(sqlFile))) {
            //REPLACE INTO (.....) VALUES
            String header = reader.readLine() + reader.readLine();
            String tmp;
            List<String> values = new ArrayList<>(Constants.BATCH_SIZE);
            while (StringUtils.isNotBlank((tmp = reader.readLine()))) {
                values.add(tmp);
                if (values.size() == Constants.BATCH_SIZE) {
                    index = executeRecover(recoverTask.getId(), index, totalCount, header, values);
                }
            }
            if (values.size() > 0) {
                executeRecover(recoverTask.getId(), index, totalCount, header, values);
            }
        }
    }

    private long executeRecover(Long taskId, Long index, Long totalCount, String sqlHeader, List<String> values) {
        index += values.size();

        String sql = sqlHeader + StringUtils.join(values, "");
        if (sql.endsWith(Constants.COMMA)) {
            sql = sql.substring(0, sql.length() - 1);
        }
        baseMapper.executeRecover(sql);
        values.clear();

        baseMapper.updateFinishInfo(taskId, ArchiveJobExecuteStatusEnum.EXECUTING.getValue(),
            ArchiveUtil.getInfo(String.format("恢复进度 %d/%d", index, totalCount)));

        return index;
    }

    private void recoverFromDb(ArchiveRecoverTask recoverTask) throws Exception {
        ArchiveStrategy strategy = recoverTask.getArchiveStrategy();
        ArchiveResult result = recoverTask.getArchiveResult();
        ArchiveLocation location = recoverTask.getArchiveLocation();

        List<String> fromCells = Arrays.asList(strategy.getTargetFields().split(Constants.COMMA));
        List<String> toCells = Arrays.asList(strategy.getSourceFields().split(Constants.COMMA));
        String minPk = result.getMinPk();
        String maxPk = result.getMaxPk();
        long totalCount = result.getTotalCount();

        long index = 0;
        try (Connection connection = ArchiveUtil.getDbConnection(location)) {
            String tmpSql = ArchiveUtil.getSelectSql(strategy.getTargetTable(), fromCells, strategy.getTargetPk(), minPk);
            List<Map<String, Object>> temp = ArchiveUtil.queryDb(connection, tmpSql, fromCells);
            if (temp.isEmpty()) {
                throw new ServiceException(String.format("%s：恢复失败，要恢复的数据在备份库中不存在", DateUtil.formatDateTime(new Date())));
            }
            String sqlHeader = ArchiveUtil.getInsertHeadSql(strategy.getSourceTable(), toCells);
            List<String> sqlValues = ArchiveUtil.getDataSqlLineList(temp, toCells);
            //插入数据库
            index = executeRecover(recoverTask.getId(), index, totalCount, sqlHeader, sqlValues);
            minPk = temp.get(0).get(strategy.getTargetPk()).toString();
            while (ArchiveUtil.pkCompareTo(minPk, maxPk) < 0) {
                tmpSql = ArchiveUtil.getSelectSql(strategy.getTargetTable(), fromCells, strategy.getTargetPk(), minPk, maxPk);
                if (CollectionUtils.isNotEmpty(temp = ArchiveUtil.queryDb(connection, tmpSql, fromCells))) {
                    minPk = temp.get(temp.size() - 1).get(strategy.getSourcePk()).toString();
                    sqlValues = ArchiveUtil.getDataSqlLineList(temp, toCells);
                    index = executeRecover(recoverTask.getId(), index, totalCount, sqlHeader, sqlValues);
                } else {
                    break;
                }
            }
        }
    }

    private void recoverCheck(ArchiveRecoverTask recoverTask) {
        ArchiveResult archiveResult = resultMapper.selectById(recoverTask.getResultId());
        if (archiveResult == null || ArchiveJobExecuteStatusEnum.SUCCESS.getValue() != archiveResult.getExecuteStatus()) {
            Date endTime = new Date();
            if (archiveResult == null) {
                throw new ServiceException(String.format("%s：恢复失败，要恢复的归档结果不存在", DateUtil.formatDateTime(endTime)));
            } else {
                throw new ServiceException(String.format("%s：恢复失败，要恢复的归档结果不是成功状态", DateUtil.formatDateTime(endTime)));
            }
        }
        ArchiveTask archiveTask = archiveTaskMapper.selectById(archiveResult.getTaskId());
        if (archiveTask == null) {
            throw new ServiceException(String.format("%s：恢复失败，要恢复的归档任务不存在", DateUtil.formatDateTime(new Date())));
        }
        ArchiveStrategy archiveStrategy = strategyMapper.selectById(archiveTask.getStrategyId());
        if (archiveStrategy == null) {
            throw new ServiceException(String.format("%s：恢复失败，要恢复的归档任务的归档策略不存在", DateUtil.formatDateTime(new Date())));
        }
        ArchiveLocation archiveLocation = locationMapper.selectById(archiveStrategy.getLocationId());
        if (archiveLocation == null) {
            throw new ServiceException(String.format("%s：恢复失败，要恢复的归档任务的归档策略的归档位置不存在", DateUtil.formatDateTime(new Date())));
        }
        recoverTask.setArchiveTask(archiveTask);
        recoverTask.setArchiveResult(archiveResult);
        recoverTask.setArchiveStrategy(archiveStrategy);
        recoverTask.setArchiveLocation(archiveLocation);
    }
}
